import java.io.File
import java.util.Date

import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * t_acc_balance 导入对账数据
  */
object testTen {

  def main(args: Array[String]) {
    val warehouseLocation = new File("hdfs://yun0:9000/hive/warehouse/").getAbsolutePath

    val spark = SparkSession
      .builder()
      .appName("Spark Hive Example")
      .config("spark.sql.warehouse.dir", warehouseLocation)
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._
    import spark.sql


    var now = new Date()
    val time = now.getTime
    var sqlStr :String = "INSERT into test.t_acc_balance "+
      "select '20150815' as stat_date, "+
      "tt1.facct_type as facct_type, " +
      "if ( begin_remain is NULL, 0, begin_remain ) as begin_remain, " +
      "if ( end_remain is NULL, 0, end_remain ) as end_remain, "+
      "if ( charge_amt is NULL, 0, charge_amt ) as charge_amt, " +
      "if ( consume_amt is NULL, 0, consume_amt ) as consume_amt, "+
      " '0' as gift_flag, "+
      time + " as fchg_time, " +
      "tt1.fsett_id as fsett_id "+
      "from ( "+
      "select facct_type, sum(fbalance-fgen_balance) as end_remain, fsett_id "+
      "from test.t_imt_info_uniacct_day t1 "+
      "where fstatus=1 and fis_test=0 and fsett_id in ('caccts','shanghaidq','xianling','hk','gdt','jianling','jinquan','tqyxjf') "+
      "group by facct_type, fsett_id "+
      ") tt1 "+
      "left join "+
      "( "+
      "select facct_type, sum(fbalance-fgen_balance) as begin_remain, fsett_id "+
      "from test.t_imt_info_uniacct_day t2 "+
      "where fstatus=1 and fis_test=0 and fsett_id in ('caccts','shanghaidq','xianling','hk','gdt','jianling','jinquan','tqyxjf') "+
      "group by facct_type,fsett_id "+
      ") tt2 "+
      "on tt1.facct_type = tt2.facct_type "+
      "left JOIN "+
      "( "+
      "select facct_type,fsett_id, "+
      "sum(charge_amt) as charge_amt, "+
      "sum(consume_amt) as consume_amt "+
      "from ( "+
      "select facct_type,fsett_id, "+
      "case ftrans_type "+
      "when '11' then fpay_amt-fgen_amt "+
      "when '12' then fpay_amt-fgen_amt " +
      "when '13' then fpay_amt-fgen_amt "+
      "when '14' then fpay_amt-fgen_amt " +
      "when '15' then fpay_amt-fgen_amt " +
      "when '16' then fpay_amt-fgen_amt " +
      "when '23' then fgen_amt-fpay_amt " +
      "when '24' then fgen_amt-fpay_amt " +
      "when '17' then fgen_amt-fpay_amt "+
      "else 0.0  end as consume_amt, " +
      "case ftrans_type " +
      "when '21' then fpay_amt-fgen_amt " +
      "when '22' then fpay_amt-fgen_amt " +
      "when '25' then fpay_amt-fgen_amt " +
      "when '26' then fpay_amt-fgen_amt " +
      "when '27' then fpay_amt-fgen_amt " +
      "when '28' then fpay_amt-fgen_amt " +
      "when '51' then fgen_amt-fpay_amt " +
      "else 0.0 end as charge_amt " +
      "from test.t_imt_pay_uniacct_day t3 " +
      "where fis_test=0 and fsett_id in ('caccts','shanghaidq','xianling','hk','gdt','jianling','jinquan','tqyxjf') ) t4 "+
      "group by facct_type,fsett_id ) tt3 "+
      "on tt1.facct_type = tt3.facct_type and tt1.fsett_id=tt3.fsett_id"




    var sqlStr1 : String = "INSERT INTO TABLE tendb.t_acc_balance SELECT '20170815' AS stat_date,stt1.facct_type AS facct_type ,IF ( begin_remain IS NULL ,0 ,begin_remain ) AS begin_remain ,IF ( end_remain IS NULL ,0 ,end_remain ) AS end_remain ,IF ( charge_amt IS NULL ,0 ,charge_amt ) AS charge_amt ,IF ( consume_amt IS NULL ,0 ,consume_amt ) AS consume_amt ,'0' AS gift_flag ,'' AS fchg_time ,stt1.fsett_id AS fsett_id FROM (  SELECT facct_type ,SUM(fbalance - fgen_balance) AS end_remain ,fsett_id FROM tendb.t_imt_info_uniacct_day  t1 WHERE fstatus = 1 AND fis_test = 0  AND fsett_id IN ( 'caccts' ,'shanghaidq' ,'xianling' ,'hk' ,'gdt' ,'jianling' ,'jinquan' ,'tqyxjf' )  GROUP BY facct_type ,fsett_id ) stt1 LEFT JOIN ( SELECT facct_type ,SUM(fbalance - fgen_balance) AS begin_remain ,fsett_id FROM tendb.t_imt_info_uniacct_day t2 WHERE fstatus = 1 AND fis_test = 0 AND fsett_id IN ( 'caccts' ,'shanghaidq'  ,'xianling' ,'hk' ,'gdt' ,'jianling' ,'jinquan' ,'tqyxjf' ) GROUP BY facct_type ,fsett_id ) stt2  ON stt1.facct_type = stt2.facct_type LEFT JOIN ( SELECT facct_type ,fsett_id ,SUM(charge_amt) AS charge_amt ,SUM(consume_amt) AS consume_amt  FROM ( SELECT facct_type ,fsett_id ,CASE ftrans_type WHEN '11' THEN fpay_amt - fgen_amt  WHEN '12' THEN fpay_amt - fgen_amt WHEN '13' THEN fpay_amt - fgen_amt WHEN '14' THEN fpay_amt - fgen_amt WHEN '15' THEN fpay_amt - fgen_amt WHEN '16' THEN fpay_amt - fgen_amt  WHEN '23' THEN fgen_amt - fpay_amt WHEN '24' THEN fgen_amt - fpay_amt WHEN '17' THEN fgen_amt - fpay_amt ELSE 0.0 END AS consume_amt ,CASE ftrans_type  WHEN '21' THEN fpay_amt - fgen_amt WHEN '22' THEN fpay_amt - fgen_amt WHEN '25' THEN fpay_amt - fgen_amt  WHEN '26' THEN fpay_amt - fgen_amt WHEN '27' THEN fpay_amt - fgen_amt  WHEN '28' THEN fpay_amt - fgen_amt WHEN '51' THEN fgen_amt - fpay_amt  ELSE 0.0 END AS charge_amt FROM tendb.t_imt_pay_uniacct_day t3 WHERE fis_test = 0  AND fsett_id IN ( 'caccts' ,'shanghaidq' ,'xianling' ,'hk' ,'gdt' ,'jianling' ,'jinquan' ,'tqyxjf' ) ) t4 GROUP BY facct_type ,fsett_id ) stt3 ON stt1.facct_type = stt3.facct_type AND stt1.fsett_id = stt3.fsett_id";

    var sqlstr3 : String = "SELECT * FROM ( SELECT t1.statis_date AS statis_date ,t1.fuin AS acct_id ,t1.facct_type AS acct_type ,'qqpoint' AS channel ,'' AS subchannel ,'info' AS datatype ,CASE WHEN t2.fbalance IS NULL THEN - CAST(t1.fbalance AS int) ELSE t2.fbalance - CAST(t1.fbalance AS int) END AS amt ,t1.fopen_time AS open_time ,t1.fchg_time AS chg_time ,t1.flast_tran_time AS last_tran_time ,t1.fstatus AS status ,'' as remark FROM ( SELECT * FROM test.t_int_qpoint_acct_info WHERE statis_date = 20170813 AND facct_type = '3' ) t1 LEFT JOIN ( SELECT * FROM test.t_int_qpoint_acct_info WHERE statis_date = 20170813 AND facct_type = '3' ) t2 ON ( t1.fuin = t2.fuin AND t1.facct_type = t2.facct_type ) WHERE CASE WHEN t2.fbalance IS NULL THEN - CAST(t1.fbalance AS int) ELSE t2.fbalance - CAST(t1.fbalance AS int) END != 0 UNION ALL SELECT 20170813 AS statis_date ,NVL(t3.fuin, t4.fuin) AS facct_id ,NVL(t3.facct_type, t4.facct_type) AS facct_type ,'qqpoint' ,'' ,'water' ,CASE WHEN t3.out_amt IS NULL THEN - t4.in_amt WHEN t4.in_amt IS NULL THEN t3.out_amt ELSE t3.out_amt - t4.in_amt END AS amt ,t3.out_amt ,t4.in_amt ,'' ,CASE WHEN out_num IS null THEN in_num WHEN in_num IS null THEN out_num ELSE out_num + in_num END ,'' FROM ( SELECT fuin ,facct_type ,SUM(ftran_amt) AS out_amt ,COUNT(ftran_amt) AS out_num FROM ( SELECT ta3.* ,ta4.fchg_time AS beg_time FROM ( SELECT ta1.* ,ta2.fchg_time AS end_time FROM ( SELECT * FROM test.t_int_qpoint_acct_water WHERE statis_day = 20170813 AND facct_type = '3' ) ta1 LEFT JOIN ( SELECT * FROM test.t_int_qpoint_acct_info WHERE statis_date = 20170813 ) ta2 ON ( ta1.fuin = ta2.fuin AND ta1.facct_type = ta2.facct_type ) ) ta3 LEFT JOIN ( SELECT * FROM test.t_int_qpoint_acct_info WHERE statis_date = 20170813 ) ta4 ON ( ta3.fuin = ta4.fuin AND ta3.facct_type = ta4.facct_type ) ) WHERE fio_flag = 1 AND end_time IS NOT NULL AND ftran_type <> 101 AND ftran_type <> 102 AND UNIX_TIMESTAMP(end_time) >= UNIX_TIMESTAMP(fchg_time) AND ( beg_time IS null OR UNIX_TIMESTAMP(beg_time) < UNIX_TIMESTAMP(fchg_time) ) GROUP BY fuin ,facct_type ) t3 FULL JOIN ( SELECT fuin ,facct_type ,SUM(ftran_amt) AS in_amt ,COUNT(ftran_amt) AS in_num FROM ( SELECT ta3.* ,ta4.fchg_time AS beg_time FROM ( SELECT ta1.* ,ta2.fchg_time AS end_time FROM ( SELECT * FROM test.t_int_qpoint_acct_water WHERE statis_day = 20170813 AND facct_type = '3' ) ta1 LEFT JOIN ( SELECT * FROM test.t_int_qpoint_acct_info WHERE statis_date = 20170813 ) ta2 ON ( ta1.fuin = ta2.fuin AND ta1.facct_type = ta2.facct_type ) ) ta3 LEFT JOIN ( SELECT * FROM test.t_int_qpoint_acct_info WHERE statis_date = 20170813 ) ta4 ON ( ta3.fuin = ta4.fuin AND ta3.facct_type = ta4.facct_type ) ) WHERE fio_flag = 2 AND end_time IS NOT NULL AND ftran_type <> 101 AND ftran_type <> 102 AND UNIX_TIMESTAMP(end_time) >= UNIX_TIMESTAMP(fchg_time) AND ( beg_time IS null OR UNIX_TIMESTAMP(beg_time) < UNIX_TIMESTAMP(fchg_time) ) GROUP BY fuin ,facct_type ) t4 ON ( t3.fuin = t4.fuin AND t3.facct_type = t4.facct_type ) WHERE CASE WHEN t3.out_amt IS NULL THEN - t4.in_amt WHEN t4.in_amt IS NULL THEN t3.out_amt ELSE t3.out_amt - t4.in_amt END != 0 )"

//    var dataFrame  = sql(sqlstr3)
//    dataFrame.queryExecution.
//    //print(dataFrame.queryExecution.logical)
////    dataFrame.write.save("hdfs://yun0:9000/test/qpoing_balance.csv")
////    dataFrame.write.csv("/data/tendata/qpoing_balance2.csv")
//
//    // sql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING) USING test")
//    //    sql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src")
//
//
//    // Queries are expressed in HiveQL
//    sql("show databases").show()
//    //sql("show tables using test").show()
//    sql("SELECT * FROM test.t_imt_info_uniacct_day ").show()
//    //    runBasicParquetExample(spark)

    //    val output = count.saveAsTextFile("/home/mark/testfile/helloSparkRes")
  }

}
